package com.atguigu.flink.streamapi.trans;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/1/13
 * 
 *      在sql语句中，必须先group by 再聚合。
 *      
 *      在flink，分组使用keyBy，之后才能聚合。
 *              聚合的功能都在KeyedStream中定义！
 */
public class Demo6_Agg
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //SingleOutputStreamOperator<T> extends DataStream
        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());


        //KeyedStream<T, KEY> extends DataStream<T>
        KeyedStream<WaterSensor, String> ds2 = ds.keyBy(WaterSensor::getId);

        /*
                select
                    id,max(vc), 常量
                from xx
                group by id



                max(String field):  所求数据的除了分组和聚合的字段外，其他字段默认取当前组中第一条的属性。
                maxBy(String field, boolean first): 求vc最大的那条数据的所有属性
                        boolean first： 如果出现了两条数据的vc都是最大，取哪条?
                                boolean first = true ，取第一条
                                boolean first = false，取最后一条
         */
        ds2
            .maxBy("vc",false)
            //.minBy()
            //.max("vc")
            //.min("vc")
            //.sum("vc")
           .print();




        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
}
